Let's remove SQS from this and make it more generic in naming.

Accepted

Blocker — Vendor neutrality: This mixin is part of the core backend abstraction layer, but it leaks AWS-specific concepts. The class name is fine, but start_sqs_consumer names a specific AWS service, and the docstrings reference SQS throughout.

For an Apache project, this should be vendor-neutral:

• Rename start_sqs_consumer() → start_event_consumer()
• Update docstrings to describe generic event-driven updates, not SQS specifically
• The is_event_driven() method name is already good

This would allow a GCP Pub/Sub or Azure Event Grid implementation to use the same mixin without the naming being misleading.

Accepted

Medium — Use an Enum instead of magic strings: "POLLING" and "SQS" are magic strings that are error-prone (e.g., typo "Polling" would silently fall through). Consider:

from enum import Enum

class TrackingMode(str, Enum):
    POLLING = "POLLING"
    EVENT_DRIVEN = "EVENT_DRIVEN"

Also, "SQS" is an AWS-specific name. Since this setting is in the S3 backend (already AWS-specific), it's less critical than the core mixin, but "EVENT_DRIVEN" would still be more descriptive of the behavior rather than the implementation.

High — No graceful shutdown: This while True loop runs as a fire-and-forget asyncio.create_task() (see run.py). When the server shuts down, this task will be cancelled, but asyncio.CancelledError is not handled. The outer except Exception won't catch it (in Python 3.9+, CancelledError inherits from BaseException).

Recommendation:

try:
    while True:
        ...
except asyncio.CancelledError:
    logger.info("SQS consumer shutting down")
    raise

Also, the task reference in run.py is not stored anywhere — it could be garbage collected. Store it and cancel it in the shutdown path of the lifespan context manager.

Medium — Silent error swallowing: Catching Exception and only logging means data loss can go completely unnoticed. If indexing fails for a file, that tracking data is permanently skipped.

Consider either:

1. Re-raising after logging (letting the message go back to SQS for retry, eventually to DLQ)
2. At minimum, emitting a metric/counter so operators can alert on indexing failures

Currently the message gets deleted from SQS even if _handle_s3_event fails (the delete_message call is outside this try/except), so failed events are lost forever.

@vaquarkhan bump on this.

High — Code duplication: BedrockStreamingAction duplicates ~90% of BedrockAction:

• Identical __init__ (all parameters, client creation, instance variable assignments)
• Identical reads, writes, name properties
• Identical request-building logic in stream_run vs run_and_update

Consider extracting a shared base class:

class _BedrockBase:
    def __init__(self, model_id, input_mapper, reads, writes, ...):
        # shared init
    
    def _build_request(self, state: State) -> Dict[str, Any]:
        # shared request building

class BedrockAction(_BedrockBase, SingleStepAction): ...
class BedrockStreamingAction(_BedrockBase, StreamingAction): ...

and also move this to a different PR

Medium — Client created at construction time: The boto3 client is created eagerly in __init__, which means:

1. The action can't be serialized/pickled (relevant for distributed execution)
2. Testing requires mocking at construction time or having valid AWS credentials
3. Credentials are captured at construction, not at invocation

Consider lazy client creation (create on first run_and_update call) or accepting a client/session as a parameter for testability.

I think any client should be passed in (i.e. injected) -- that way it's easier for tests etc.

High — Task reference not stored: asyncio.create_task() returns a task reference that isn't assigned to anything. This means:

1. The task could be garbage collected
2. There's no way to cancel it during shutdown
3. Exceptions in the task will be silently lost

Store the task and cancel it in the shutdown path:

sqs_task = None
if isinstance(backend, EventDrivenBackendMixin) and backend.is_event_driven():
    sqs_task = asyncio.create_task(backend.start_event_consumer())
yield
if sqs_task:
    sqs_task.cancel()
    try:
        await sqs_task
    except asyncio.CancelledError:
        pass

Also, import asyncio should be a top-level import, not inside the function.

Blocker — Should not be committed: .terraform.lock.hcl is a generated file specific to the contributor's Terraform version and platform. It should be added to .gitignore (or the terraform .gitignore), not committed to the repo.

More broadly: the entire terraform/ directory (including dev.tfvars, prod.tfvars with hardcoded us-east-1) should live under examples/deployment/aws/ or contrib/aws/ rather than at the repo root. Burr is a Python framework library — shipping infrastructure-as-code at the top level makes the project appear AWS-first rather than vendor-neutral.

Medium — Tests are no-ops without boto3: Every test is wrapped in try/except ImportError with an assertion on the error message. When boto3 isn't installed (which is the case in CI unless explicitly added), all tests silently pass without testing anything.

Better approaches:

1. Use pytest.importorskip("boto3") to clearly skip tests when the dependency is missing
2. Mock boto3 for unit tests so they run regardless of whether boto3 is installed
3. For integration tests that need real AWS, mark them with @pytest.mark.integration and skip in CI

When this moves to a separate PR, you can follow what we do for the databases plugins and have explicit integration tests -- so we'd just remove the imports

move to separate PR

I don't think we need this?

let's move these to an aws folder under terraform or something

all are fixed and in PR

can you move these to the top level please? we can then add pytest.importorskip if the dependencies aren't installed.